package ProcessFunctionTest

import Source.SensorReading
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * 侧输出流
 */
object SidOutPutTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val inputStream = env.socketTextStream("localhost", 7777)

    //转换成样例类
    val dataStream = inputStream
      .map(data => {
        val arr = data
          .split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      }).keyBy(_.id)

    //分流
    val highTempStream = dataStream
      .process(new SplitTempProcessor(30.0))

    highTempStream.print("high")
    //获取侧输出流，测输出流的输出类型和主流不一定一致,这边的输入需要和下面定义的侧输出流一致
    highTempStream.getSideOutput(new OutputTag[(String, Long, Double)]("low")).print("low")

    env.execute()
  }
}

/**
 * 自定义ProcessFunction，来进行分流
 * 这边定义的输出类型时主流的输出类型
 *
 * @param threshold
 */
class SplitTempProcessor(threshold: Double) extends ProcessFunction[SensorReading, SensorReading] {
  override def processElement(i: SensorReading, context: ProcessFunction[SensorReading, SensorReading]#Context,
                              collector: Collector[SensorReading]): Unit = {
    if (i.temperature > 30) {
      //高温流,输出到主流
      collector.collect(i)
    } else {
      //输出到测输出流
      context.output(new OutputTag[(String, Long, Double)]("low"), (i.id, i.timeStamp, i.temperature))
    }
  }
}